/*
* Copyright (c) 2025 Shenzhen Kaihong Digital Industry Development Co., Ltd.
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

use std::{net::{IpAddr, SocketAddr}, process::Command, time::Duration};

use communication_layer_request_reply::{RequestReplyLayer, TcpLayer, TcpRequestReplyConnection};
use dora_daemon::Daemon;
use dora_message::{cli_to_coordinator::ControlRequest, coordinator_to_cli::ControlRequestReply};
use dora_node_api::dora_core::topics::{DORA_COORDINATOR_PORT_CONTROL_DEFAULT, LOCALHOST};
use eyre::{Context, ContextCompat};

pub struct DoraServer;

impl DoraServer {
    pub fn start() -> eyre::Result<()> {
        let mut session = match connect_to_coordinator(Self::default_coordinator_addr()) {
            Ok(session) => session,
            Err(_) => {
                Self::run_self("coordinator")?;
    
                loop {
                    match connect_to_coordinator(Self::default_coordinator_addr()) {
                        Ok(session) => break session,
                        Err(_) => {
                            // sleep a bit until the coordinator accepts connections
                            std::thread::sleep(Duration::from_millis(50));
                        }
                    }
                }
            }
        };

        if !daemon_running(&mut *session)? {
            Self::run_self("daemon")?;
            
            // wait a bit until daemon is connected
            let mut i = 0;
            const WAIT_S: f32 = 0.1;
            loop {
                if daemon_running(&mut *session)? {
                    break;
                }
                i += 1;
                if i > 20 {
                    eyre::bail!("daemon not connected after {}s", WAIT_S * i as f32);
                }
                std::thread::sleep(Duration::from_secs_f32(WAIT_S));
            }
        }

        Ok(())
    }

    pub fn stop() -> eyre::Result<()> {
        match connect_to_coordinator(Self::default_coordinator_addr()) {
            Ok(mut session) => {
                // send destroy command to dora-coordinator
                let reply_raw = session
                    .request(&serde_json::to_vec(&ControlRequest::Destroy).unwrap())
                    .wrap_err("failed to send destroy message")?;
                let result: ControlRequestReply =
                    serde_json::from_slice(&reply_raw).wrap_err("failed to parse reply")?;
                match result {
                    ControlRequestReply::DestroyOk => {
                        tracing::info!("Coordinator and daemons destroyed successfully");
                    }
                    ControlRequestReply::Error(err) => {
                        return Err(eyre::anyhow!("Destroy command failed with error: {}", err));
                    }
                    _ => {
                        return Err(eyre::anyhow!("Unexpected reply from dora-coordinator"));
                    }
                }
            }
            Err(_) => {
                return Err(eyre::anyhow!("Could not connect to dora-coordinator"));
            }
        }

        Ok(())
    }

    pub fn is_running() -> bool {
        let Ok(mut session) = connect_to_coordinator(Self::default_coordinator_addr()) else {
            return false;
        };
        daemon_running(&mut *session).is_ok_and(|running| running)
    }

    pub async fn start_coordinator(
        interface: IpAddr,
        port: u16,
        control_interface: IpAddr,
        control_port: u16,
    ) -> eyre::Result<()> {
        let bind = SocketAddr::new(interface, port);
        let bind_control = SocketAddr::new(control_interface, control_port);
        let (port, task) = dora_coordinator::start(bind, bind_control, futures::stream::empty()).await?;
        tracing::info!("Listening for incoming daemon connection on {port}");
        task.await.context("failed to start dora-coordinator")?;
        Ok(())
    }

    pub async fn start_daemon(
        local_listen_port: u16,
        coordinator_addr: IpAddr,
        coordinator_port: u16,
    ) -> eyre::Result<()> {
        Daemon::run(SocketAddr::new(coordinator_addr, coordinator_port), None, local_listen_port)
            .await.context("failed to start dora-daemon")
    }

    fn default_coordinator_addr() -> SocketAddr {
        (LOCALHOST, DORA_COORDINATOR_PORT_CONTROL_DEFAULT).into()
    }

    fn run_self(arg: &str) -> eyre::Result<()> {
        let program = std::env::args_os()
            .next()
            .context("Could not get ohmw_dora_server path")?;
        Command::new(program)
            .arg(arg)
            .spawn()
            .wrap_err(format!("failed to run ohmw_dora_server '{arg}'"))?;
        Ok(())
    }
}

fn connect_to_coordinator(coordinator_addr: SocketAddr) -> std::io::Result<Box<TcpRequestReplyConnection>> {
    TcpLayer::new().connect(coordinator_addr)
}

fn daemon_running(session: &mut TcpRequestReplyConnection) -> Result<bool, eyre::ErrReport> {
    let reply_raw = session
        .request(&serde_json::to_vec(&ControlRequest::DaemonConnected).unwrap())
        .wrap_err("failed to send DaemonConnected message")?;

    let reply = serde_json::from_slice(&reply_raw).wrap_err("failed to parse reply")?;
    let running = match reply {
        ControlRequestReply::DaemonConnected(running) => running,
        other => return Err(eyre::anyhow!("unexpected reply to daemon connection check: {other:?}")),
    };

    Ok(running)
}
